package org.lql.chaos.kafka.example.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 01:39
 * @description: 自定义消费者拦截器
 **/
@Slf4j
public class ExampleConsumerInterceptor implements ConsumerInterceptor<String, Object> {

	@Override
	public ConsumerRecords<String, Object> onConsume(ConsumerRecords<String, Object> records) {
		Map<TopicPartition, List<ConsumerRecord<String, Object>>> newRecords = new HashMap<>();
		for (TopicPartition tp : records.partitions()) {
			List<ConsumerRecord<String, Object>> filtered = records.records(tp)
					.stream()
					.filter(r -> r.value() != null)
					.collect(Collectors.toList());
			newRecords.put(tp, filtered);
		}
		return new ConsumerRecords<>(newRecords);
	}

	@Override
	public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {
		log.info("onCommit: {}", map);
	}

	@Override
	public void close() {

	}

	@Override
	public void configure(Map<String, ?> map) {

	}
}
